package com.spark.core.transformation

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
 * mapPartitions 遍历的是每个分区中的数据，一个个分区的遍历
 * 每个分组触发一次函数的执行, input Iterator; output Iterator
 * 相对于map 一条条处理数据，性能比较高。
 */
object Demo17_mapPartitions {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("mapPartitions")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    val infos = sc.parallelize(List[String]("a", "b", "c", "d", "e", "f", "g"), 2)

    /*
    mapPartitions 每次处理的是一个分区的数据,
    返回集合的迭代器, 集合的泛型就是RDD的泛型
     */
    val result: RDD[String] = infos.mapPartitions(iter => {
      val array = new ArrayBuffer[String]()
      array.append("创建数据库连接... ...")
      iter.foreach(elem => {
        array.append("执行sql----" + elem)
      })
      array.append("关闭数据库连接... ...")
      array.iterator
    })

    println("---------------")
    result.foreach(println)

  }
}
